package com.example.streamsqlpark.entity;

import com.baomidou.mybatisplus.annotation.*;
import com.example.streamsqlpark.base.util.JacksonUtil;
import com.example.streamsqlpark.base.util.YarnQueueLabelExpression;
import com.example.streamsqlpark.common.conf.ConfigConst;
import com.example.streamsqlpark.common.util.HttpClientUtils;
import com.example.streamsqlpark.enums.ClusterState;
import com.example.streamsqlpark.enums.ExecutionMode;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import lombok.Data;
import lombok.SneakyThrows;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.client.config.RequestConfig;

import java.io.Serializable;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.*;

/**
 * @author voloiono
 * @date 2024/4/17 9:49
 */
@Data
@TableName("t_flink_cluster")
public class FlinkCluster implements Serializable {
    @TableId(type = IdType.AUTO)
    private Long id;

    @TableField(updateStrategy = FieldStrategy.IGNORED)
    private String address;

    private String clusterId;

    private String clusterName;

    private Integer executionMode;

    /**
     * flink version
     */
    private Long versionId;

    private String k8sNamespace;

    private String serviceAccount;

    private String description;

    private Long userId;

    private String flinkImage;

    private String options;

    private String yarnQueue;

    private Boolean k8sHadoopIntegration;

    private String dynamicProperties;

    private Integer k8sRestExposedType;

    private String k8sConf;

    private Integer resolveOrder;

    @TableField(updateStrategy = FieldStrategy.IGNORED)
    private String exception;

    private Integer clusterState;

    private Date createTime = new Date();

    public ExecutionMode getExecutionMode() {
        return ExecutionMode.of(this.executionMode);
    }

    public ClusterState getClusterState() {
        return ClusterState.of(this.clusterState);
    }

    @JsonIgnore
    @SneakyThrows
    public Map<String, Object> getOptionMap() {
        if (StringUtils.isBlank(this.options)) {
            return Collections.emptyMap();
        }
        Map<String, Object> map = JacksonUtil.read(this.options, Map.class);
        if (ExecutionMode.YARN_SESSION.equals(getExecutionMode())) {
            map.put(ConfigConst.KEY_YARN_APP_NAME(), this.clusterName);
            map.putAll(YarnQueueLabelExpression.getQueueLabelMap(yarnQueue));
        }
        map.entrySet().removeIf(entry -> entry.getValue() == null);
        return map;
    }

    @JsonIgnore
    public URI getRemoteURI() {
        try {
            HttpClientUtils.httpGetRequest(
                    this.address, RequestConfig.custom().setSocketTimeout(2000).build()
            );
            return new URI(address);
        } catch (URISyntaxException e) {
        }
        return null;
    }

    public boolean verifyClusterConnection(){
        if (ExecutionMode.REMOTE.equals(this.getExecutionMode())){
            if (address==null){
                return false;
            }

            try {
            String resUrl=address+"/overview";
            String result = HttpClientUtils.httpGetRequest(
                    resUrl, RequestConfig.custom().setSocketTimeout(2000).build()
            );
                JacksonUtil.read(result, Overview.class);
                return true;
            } catch (JsonProcessingException e) {
            }
            return false;
        }
        return false;
    }

    public Map<String,String> getFlinkConfig() throws JsonProcessingException {
        String confUrl=this.address+"/jobmanager/config";
        String json = HttpClientUtils.httpGetRequest(
                confUrl, RequestConfig.custom().setSocketTimeout(2000).build()
        );
        if (StringUtils.isEmpty(json)){
            return Collections.emptyMap();
        }
        List<Map<String, String>> configList = JacksonUtil.read(json, new TypeReference<List<Map<String, String>>>() {
        });
        HashMap<String, String> config = new HashMap<>(0);
        configList.forEach(k->config.put(k.get("key"),k.get("value")));
        return config;
    }
}
